package org.suyafei.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple9;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.suyafei.model.ShortLinkVisitStatsDO;
import org.suyafei.util.KafkaUtil;
import org.suyafei.util.TimeUtil;
import redis.clients.jedis.Tuple;

import java.time.Duration;

/**
 * @desc:
 * @author:yafei.su
 * @date: 2022/5/10
 */
public class DwsShortLinkVisitStatsApp {

    /**
     * 宽表
     */
    public static final String SHORT_LINK_SOURCE_TOPIC = "dwm_link_visit_topic";

    public static final String SHORT_LINK_SOURCE_GROUP = "dws_link_visit_group";

    /**
     * uv的数据流
     */
    public static final String UNIQUE_VISITOR_SOURCE_TOPIC = "dwm_unique_visitor_topic";

    public static final String UNIQUE_VISITOR_SOURCE_GROUP = "dws_unique_visitor_group";

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);


        //1、获取多个数据
        DataStreamSource<String> shortLinkDS = env.addSource(KafkaUtil.getKafkaConsumer(SHORT_LINK_SOURCE_TOPIC, SHORT_LINK_SOURCE_GROUP));
        DataStreamSource<String> uniqueVisitorDS = env.addSource(KafkaUtil.getKafkaConsumer(UNIQUE_VISITOR_SOURCE_TOPIC, UNIQUE_VISITOR_SOURCE_GROUP));

        //2、结构转换 uniqueVisitorDS、shortLinkDS
        SingleOutputStreamOperator<ShortLinkVisitStatsDO> shortLinkMapDS = shortLinkDS.map(new MapFunction<String, ShortLinkVisitStatsDO>() {
            @Override
            public ShortLinkVisitStatsDO map(String s) throws Exception {
                ShortLinkVisitStatsDO shortLinkVisitStatsDO = parseVisitStats(s);
                shortLinkVisitStatsDO.setPv(1L);
                shortLinkVisitStatsDO.setUv(0L);
                return shortLinkVisitStatsDO;
            }
        });
        SingleOutputStreamOperator<ShortLinkVisitStatsDO> uniqueVisitorMapDS = uniqueVisitorDS.map(new MapFunction<String, ShortLinkVisitStatsDO>() {
            @Override
            public ShortLinkVisitStatsDO map(String s) throws Exception {
                ShortLinkVisitStatsDO shortLinkVisitStatsDO = parseVisitStats(s);
                shortLinkVisitStatsDO.setPv(0L);
                shortLinkVisitStatsDO.setUv(1L);
                return shortLinkVisitStatsDO;
            }
        });

        //3、多流合并（合并相同结构的流）
        DataStream<ShortLinkVisitStatsDO> unionDS = shortLinkMapDS.union(uniqueVisitorMapDS);

        //4、设置WaterMark

        SingleOutputStreamOperator<ShortLinkVisitStatsDO> watermarkDS= unionDS.assignTimestampsAndWatermarks(WatermarkStrategy
                //制定允许乱序延迟最大3秒
                .<ShortLinkVisitStatsDO>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                //指定事件时间
                .withTimestampAssigner((event, timestamp) -> event.getVisitTime()));
        //5、多维度、多个字段分组
        KeyedStream<ShortLinkVisitStatsDO, Tuple9<String, String, Integer, String, String, String, String, String, String>> keyedStream = watermarkDS.keyBy(new KeySelector<ShortLinkVisitStatsDO, Tuple9<String, String, Integer, String, String, String, String, String, String>>() {
            @Override
            public Tuple9<String, String, Integer, String, String, String, String, String, String> getKey(ShortLinkVisitStatsDO shortLinkVisitStatsDO) throws Exception {
                return Tuple9.of(
                        shortLinkVisitStatsDO.getCode(),
                        shortLinkVisitStatsDO.getReferer(),
                        shortLinkVisitStatsDO.getIsNew(),
                        shortLinkVisitStatsDO.getProvince(),
                        shortLinkVisitStatsDO.getCity(),
                        shortLinkVisitStatsDO.getIp(),
                        shortLinkVisitStatsDO.getBrowserName(),
                        shortLinkVisitStatsDO.getOs(),
                        shortLinkVisitStatsDO.getDeviceType()
                );
            }
        });
        //6、开窗 15秒一次数据插入到 ck
        WindowedStream<ShortLinkVisitStatsDO, Tuple9<String, String, Integer, String, String, String, String, String, String>, TimeWindow> windowStream = keyedStream.window(TumblingEventTimeWindows.of(Time.seconds(10)));
        //7、聚合统计(补充统计起止时间)
        SingleOutputStreamOperator<Object> reduceDS= windowStream.reduce(new ReduceFunction<ShortLinkVisitStatsDO>() {
            @Override
            public ShortLinkVisitStatsDO reduce(ShortLinkVisitStatsDO shortLinkVisitStatsDO, ShortLinkVisitStatsDO t1) throws Exception {
                shortLinkVisitStatsDO.setPv(t1.getPv() + shortLinkVisitStatsDO.getPv());
                shortLinkVisitStatsDO.setUv(t1.getUv() + shortLinkVisitStatsDO.getUv());
                return shortLinkVisitStatsDO;
            }
        }, new ProcessWindowFunction<ShortLinkVisitStatsDO, Object, Tuple9<String, String, Integer, String, String, String, String, String, String>, TimeWindow>() {
            @Override
            public void process(Tuple9<String, String, Integer, String, String, String, String, String, String> tuple9, ProcessWindowFunction<ShortLinkVisitStatsDO, Object, Tuple9<String, String, Integer, String, String, String, String, String, String>, TimeWindow>.Context context, Iterable<ShortLinkVisitStatsDO> iterable, Collector<Object> collector) throws Exception {
                for (ShortLinkVisitStatsDO shortLinkVisitStatsDO : iterable) {
                    String startTime = TimeUtil.format(context.window().getStart(),TimeUtil.DEFAULT_PATTERN_WITH_TIME);
                    String endTime = TimeUtil.format(context.window().getEnd(), TimeUtil.DEFAULT_PATTERN_WITH_TIME);

                    shortLinkVisitStatsDO.setStartTime(startTime);
                    shortLinkVisitStatsDO.setEndTime(endTime);
                    collector.collect(shortLinkVisitStatsDO);
                }
            }
        });
        reduceDS.print();

        //8、输出Clickhouse


        env.execute();
    }

    private static ShortLinkVisitStatsDO parseVisitStats(String value) {

        JSONObject jsonObj = JSON.parseObject(value);

        ShortLinkVisitStatsDO visitStatsDO = ShortLinkVisitStatsDO.builder()
                .code(jsonObj.getString("code"))
                .accountNo(jsonObj.getLong("accountNo"))
                .visitTime(jsonObj.getLong("visitTime"))
                .referer(jsonObj.getString("referer"))
                .isNew(jsonObj.getInteger("isNew"))
                .udid(jsonObj.getString("udid"))

                //地理位置信息
                .province(jsonObj.getString("province"))
                .city(jsonObj.getString("city"))
                .isp(jsonObj.getString("isp"))
                .ip(jsonObj.getString("ip"))

                //设备信息
                .browserName(jsonObj.getString("browserName"))
                .os(jsonObj.getString("os"))
                .osVersion(jsonObj.getString("osVersion"))
                .deviceType(jsonObj.getString("deviceType"))

                .build();

        return visitStatsDO;
    }
}
